package m2_work;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

//工作模式---消费者
public class Consumer {

    public static void main(String[] args) throws Exception{

        //1.定义连接工厂，连接，通道
        ConnectionFactory cf = new ConnectionFactory();
        cf.setHost("192.168.64.140");
        cf.setPort(5672);
        cf.setUsername("admin");
        cf.setPassword("admin");

        Connection conn = cf.newConnection(); //要抛异常，可用Exception

        Channel ch = conn.createChannel();

        //2.定义队列queue (通过通道对象channel去定义队列，如果有叫这个名的队列，就直接中这个队列中取数据)
//        ch.queueDeclare("HelloWorld",
//                        false,
//                        false,
//                        false,
//                        null);
        //2.1 为测试“持久化”，定义一个新队列
        ch.queueDeclare("task_queue",
                        true,
                        false,
                        false,
                        null);

        System.out.println("等待接收数据");

        //如果要让mq合理地给众消费者分发数据的话，还需要设置：让消费者每次只抓取到1条消息数据。
        //在这条消息数据被处理完之前，消费者不会再去抓取新的消息数据
        // 这个设置 ，只有在 设置了让消费者手动发送消息回执之后（即ack为false时），才会生效
        //5.每次消费者只抓取1条数据
        ch.basicQos(1);

        //4.由于第3步 需要两个对象deliverCallback，cancelCallback，所以在这步要提前定义好
        //每次得到一条消息，就要把消息交给回调对象deliverCallback 去处理
        DeliverCallback deliverCallback = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                //获取到消息体本身，并把它转变成String格式
                String msg = new String(message.getBody());
                System.out.println("收到： "+msg);

                //我有一个需求：这个字符串中如果有“ . ” ，deliverCallback就等1s，然后再读取下一个字符
                //asdf....fasdf.....fa..we..
                //遍历这个msg，遇到. 就休1s
                for(int i=0; i<msg.length(); i++){
                    if(msg.charAt(i) == '.'){
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }

                //4.1 消费者要发送回执信息
                // 如果处理完了，消费者需要向mq服务器发送回执，告诉mq，消息我处理完了
                //回执信息在massage中的Envelope中的DeliveryTag中
                //最后一个参数：true 表示 如果mq服务器中还有一些 之前的一些消息 等待被确认消费者是否已经收到了的话，mq就会跟当前这条回执信息一起确认了。
                            // false就表示 mq 只确认当前这条消息的回执信息。通常 这个都设置成false
                ch.basicAck(message.getEnvelope().getDeliveryTag(),false);

                //4.2
                System.out.println("消息处理结束");
            }
        };

        CancelCallback cancelCallback = new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
                //这里暂时什么都不用写
            }
        };



        //3.消费者要 开始接收数据了 (还得通过通道对象channel调用方法接收)
        //ack -- acknowlegment  消息回执
//        ch.basicConsume("HelloWorld",
//                        false,  //关闭自动确认ack，改成手动确认ack。true的话，就是消息一旦发送，mq服务器中就删除这个消息。
//                                        //false就是mq服务器会缓存这条消息，直到收到消费者的回执信息（说明消费者已经收到了），mq才删掉这条消息
//                         deliverCallback, //由这个对象去处理接收到的msg
//                         cancelCallback);

        //3.1 消费者 从新定义的那个task_queue队列中接收数据
        ch.basicConsume("task_queue",false,deliverCallback,cancelCallback);


    }


}
